/**
 * 
 */
package edu.umd.clip.lm.playground;

import java.io.IOException;
import java.net.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

import edu.berkeley.nlp.util.*;
import edu.umd.clip.jobs.*;
import edu.umd.clip.lm.model.Experiment;
import edu.umd.clip.lm.model.LanguageModel;
import edu.umd.clip.lm.model.OnDiskCompactProbTree;
import edu.umd.clip.lm.model.ProbTree;
import edu.umd.clip.lm.storage.*;
import edu.umd.clip.lm.storage.AbstractProbTreeStorage.Key;
import edu.umd.clip.lm.storage.remote.*;

/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
public class DummyStorageClient {
	public static class Options {
        @Option(name = "-config", usage = "XML config file")
		public String config;
        @Option(name = "-jobs", usage = "number of concurrent jobs (default: 1)")
        public int jobs = 1;
        @Option(name = "-lm", required = false, usage = "LM ID to train (default: " + LanguageModel.PRIMARY_LM_ID + ")")
		public String lm = LanguageModel.PRIMARY_LM_ID;        
        @Option(name = "-host", usage = "remote storage hostname or IP address (default: use local storage)")
        public String host = null;
        @Option(name = "-port", usage = "remove storage port (default: 2332)")
        public int port = 2332;
        @Option(name = "-iterations", usage = "number of iterations (default: 10000)")
        public int iterations = 10000;
        @Option(name = "-ping", usage = "measure ping-pong latency (default: false)")
        public boolean ping = false;
	}

	/**
	 * @param args
	 * @throws UnknownHostException 
	 */
	public static void main(String[] args) throws IOException {
		// TODO Auto-generated method stub
        OptionParser optParser = new OptionParser(Options.class);
        final Options opts = (Options) optParser.parse(args, true);

		JobManager.initialize(opts.jobs);
		Thread thread = new Thread(JobManager.getInstance(), "Job Manager");
		thread.setDaemon(true);
		thread.start();

		Experiment.initialize(opts.config);
		Experiment experiment = Experiment.getInstance();
		experiment.buildPrefixes();
		experiment.buildWordPrefixes();

		InetSocketAddress address = new InetSocketAddress(InetAddress.getByName(opts.host), opts.port);
		ClientStorageConnection connection = new ClientStorageConnection(address);
		{
			Timer timer = new Timer(true);
			long interval = ClientStorageConnection.ClientMonitor.INTERVAL;
			timer.scheduleAtFixedRate(new ClientStorageConnection.ClientMonitor(connection), interval, interval);
		}

		JobManager manager = JobManager.getInstance();
		JobGroup group = manager.createJobGroup("dummies");
		
		if (opts.ping) {
			final Pinger pinger = new Pinger(connection);
			for(int i=0; i<opts.iterations; ++i) {
				manager.addJob(group, new Job(new PingJob(pinger), Integer.toString(i)));
			}		
			group.join();
			System.err.printf("Ping delay: %fms\n", Pinger.totalTime.get() * 1e-6 / Pinger.totalRequests.get());
		} else {
			AbstractProbTreeStorage storage = new ClientProbTreeStorage(connection);
			storage = new CachedProbTreeStorage(storage, 100, 200);
			
			for(int i=0; i<opts.iterations; ++i) {
				manager.addJob(group, new Job(new RetrievingJob(storage, opts.ping), Integer.toString(i)));
			}
			group.join();
			
			if (opts.ping) { 
				long totalRequests = opts.iterations * BATCH_SIZE;
				System.err.printf("Ping delay: %fms\n", RetrievingJob.pingTime.get() * 1e-6 / totalRequests);
			}
		}
	}

	private static final int BATCH_SIZE = 100;
	
	private static class PingJob implements Runnable {
		Pinger pinger;
		long start;

		public PingJob(Pinger pinger) {
			this.pinger = pinger;
		}
		
		/* (non-Javadoc)
		 * @see java.lang.Runnable#run()
		 */
		@Override
		public void run() {
			synchronized(this) {
				for(int i=0; i<BATCH_SIZE; ++i) {
					pinger.ping(this);
					try {
						wait();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		}
	}
	
	private static class Pinger implements ClientListener {
		ConcurrentHashMap<Integer, PingJob>  awaiting;
		ClientStorageConnection connection;
		static AtomicLong totalTime = new AtomicLong();
		static AtomicLong totalRequests = new AtomicLong();
		
		public Pinger(ClientStorageConnection connection) {
			this.connection = connection;
			awaiting = new ConcurrentHashMap<Integer, PingJob>(1000);
			connection.addListener(this);
		}
		
		public void ping(PingJob job) {
			PingMessage msg = new PingMessage(12);
			job.start = System.nanoTime();
			awaiting.put(msg.getId(), job);
			connection.sendMessage(msg);
		}
		/* (non-Javadoc)
		 * @see edu.umd.clip.lm.storage.remote.ClientListener#receivedReply(edu.umd.clip.lm.storage.remote.Message)
		 */
		@Override
		public void receivedReply(Message message) {
			PingJob job = awaiting.remove(message.getReplyTo());
			if (job != null) {
				synchronized(job) {
					job.notify();
				}
				long end = System.nanoTime();
				totalTime.addAndGet(end - job.start);
				totalRequests.incrementAndGet();
			}
		}

		/* (non-Javadoc)
		 * @see edu.umd.clip.lm.storage.remote.ClientListener#reconnected()
		 */
		@Override
		public void reconnected() {
		}
	}
	
	
	private static class RetrievingJob implements Runnable {
		AbstractProbTreeStorage storage;
		boolean ping;
		private static final long DELAY = 100;
		public static AtomicLong pingTime = new AtomicLong();

		public RetrievingJob(AbstractProbTreeStorage storage, boolean ping) {
			this.storage = storage;
			this.ping = ping;
		}
		
		private static void busyDelay() {
			long start = System.nanoTime();
			double num = 0.1;
			while(true) {
				num += Math.log(Math.random() + 1.0);
				if (start + DELAY * 1000000 > System.nanoTime()) break;
			}
			// prevent optimizations
			if (num > 1000000000) {
				System.out.println();
			}
		}
		/* (non-Javadoc)
		 * @see java.lang.Runnable#run()
		 */
		@Override
		public void run() {
			Random rnd = new Random();
			if (ping) {
				long start = System.nanoTime();
				for(int i=0; i<BATCH_SIZE; ++i) {
					int clusterid = rnd.nextInt();
					storage.getProbTree(0, clusterid, 0);
				}
				long end = System.nanoTime();
				pingTime.addAndGet(end - start);
			} else {
				int ids[] = new int[BATCH_SIZE];
				RequestBundle<Key,OnDiskCompactProbTree> bundle = new RequestBundle<Key,OnDiskCompactProbTree>(storage);
				
				for(int i=0; i<BATCH_SIZE; ++i) {
					int clusterid = rnd.nextInt();
					ids[i] = clusterid;
					Key key = new Key(0, clusterid, 0);
					bundle.request(key);
				}
				bundle.getResults();
			}
		}
		
	}
}
